package com.zhangh.summer.admin.support.ds.rocketmq;

import com.zhangh.summer.admin.support.ds.rocketmq.common.SendTypeEnum;
import com.zhangh.summer.admin.support.spring.util.SpringBeanUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.core.env.Environment;

import java.util.HashMap;
import java.util.Map;

/**
 * @description: TODO
 * @author: zhanghui  zhanghlucky@163.com
 * @create: 2021-07-03 22:19
 **/
@Slf4j
public class RocketMQProducer implements IRocketMQProducer {
    //public static final String DEFAULT_GROUP="default_group";
    private static final Map<String,DefaultMQProducer> PRODUCER_MAP=new HashMap<>();
    private static final Map<String,TransactionMQProducer> TRANS_PRODUCER_MAP=new HashMap<>();
    private Environment env;

    private static String nameServer;

    public RocketMQProducer(){
        this.env= SpringBeanUtil.getBeanByType(Environment.class);
        nameServer=env.getProperty("ds.rocketmq.nameServer");
    }
 /*   @Override
    public void send(String topic, String tags, String keys, String msg, SendTypeEnum sendTypeEnum) {
        this.send(DEFAULT_GROUP,topic,tags,keys,msg,sendTypeEnum);
    }*/

    @Override
    public void send(String group, String topic, String tags, String keys, String msg,SendTypeEnum sendTypeEnum) {
        DefaultMQProducer producer=getDefaultMQProducer(group);
        try {
            byte[] messageBody = msg.getBytes(RemotingHelper.DEFAULT_CHARSET);
            Message message = new Message(topic, tags, keys, messageBody);
            if (sendTypeEnum.getValue()==SendTypeEnum.SYN.getValue()){
                SendResult sendResult = producer.send(message);
                log.info("send syn message result:{}",sendResult);
            }
            else if (sendTypeEnum.getValue()==SendTypeEnum.ASYNCHRONOUS.getValue()){
                producer.send(message, new SendCallback() {
                    public void onSuccess(SendResult sendResult) {
                        log.info("send asynchronous message result:{}",sendResult);
                    }
                    public void onException(Throwable throwable) {
                        log.error("send asynchronous message error:{}",throwable.getMessage(),throwable);
                    }
                });
            }
            else if (sendTypeEnum.getValue()==SendTypeEnum.ONE_WAY.getValue()){
                producer.sendOneway(message);
                log.info("send one way message");
            }else {
                log.error("can not send message");
            }
        }catch (Exception e){
            log.error("send message error",e);
        }
    }

    @Override
    public void transSend(String group, String topic, String tags, String keys, String msg) {
        TransactionMQProducer transactionMQProducer=getTransactionMQProducer(group);
        //TODO  实现事务消息
    }

    private DefaultMQProducer getDefaultMQProducer(String group){
        DefaultMQProducer producer =PRODUCER_MAP.get(group);
        if (null!= producer){
            return producer;
        }
        synchronized (this){
            producer = new DefaultMQProducer(group);
            producer.setNamesrvAddr(nameServer);
            try {
                producer.start();
                PRODUCER_MAP.put(group, producer);
            } catch (MQClientException e) {
                e.printStackTrace();
                log.error("create DefaultMQProducer error!",e);
            }
        }
        return producer;
    }

    private  TransactionMQProducer getTransactionMQProducer(String group){
        //TransactionMQProducer producer = new TransactionMQProducer("group");
        TransactionMQProducer producer =TRANS_PRODUCER_MAP.get(group);
        if (null!= producer){
            return producer;
        }
        synchronized (this){
            producer = new TransactionMQProducer(group);
            producer.setNamesrvAddr(nameServer);
            try {
                producer.start();
                TRANS_PRODUCER_MAP.put(group, producer);
            } catch (MQClientException e) {
                e.printStackTrace();
                log.error("create TransactionMQProducer error!",e);
            }
        }
        return producer;


    }


}
